package repeater

import (
    "github.com/Shopify/sarama"
    "time"
    "../common"
)

type KafkaWriter struct {
    kafkaAddr              []string
    topic                  string
    rc              *RepeaterController
    producer               sarama.AsyncProducer
    ticker                 *time.Ticker
    partitionCount int32
}
func (ww *KafkaWriter) Init() error {
    kafkaConfig := sarama.NewConfig()
    kafkaConfig.Producer.Partitioner = sarama.NewManualPartitioner
    if ww.producer != nil {
        if err := ww.producer.Close(); err != nil {
            log.Fatal(err.Error())
        }
    }
    ww.producer = nil
    producer, err := sarama.NewAsyncProducer(ww.kafkaAddr, kafkaConfig)
    if err != nil {
        return err
    }
    ww.producer = producer
    return nil
}
func (ww *KafkaWriter) Start() {
}
func (ww *KafkaWriter) Stop() (err error) {
    return ww.producer.Close()
}

func (ww *KafkaWriter) Handle(dps *DataPointPack) error {
    for _,dp := range dps.Datapoints {
        ret:=common.Extract(dp.Name)
        if len(ret)==0{
            continue
        }
        pid := common.ExtractKey(dp.Name) % ww.partitionCount
        select {
        case ww.producer.Input() <- &sarama.ProducerMessage{Topic: ww.topic, Partition: pid, Value: dp}:
            return nil
        case err := <-ww.producer.Errors():
            return err
        }
    }
    return nil
}
func (ww *KafkaWriter) Name() string {
    return "kafka_" + ww.topic
}
func (ww *KafkaWriter) SetController(rc *RepeaterController) error {
    ww.rc = rc
    return nil
}
func (ww *KafkaWriter) Commit() (int64, error) {
    return 0, nil
}
